package cn.doitedu.data_snyc;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/1/14
 * @Desc: 学大数据，上多易教育
 *   用户注册表数据，实时同步到hbase
 **/
public class Job01_UserInfo2Hbase {
    public static void main(String[] args) {
        // 编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 建表，映射源表 : mysql 中的 ums_member
        tenv.executeSql(
                " create table ums_member_mysql(        "+
                        "     id bigint                         "+
                        "     ,member_level_id bigint           "+
                        "     ,username string                  "+
                        "     ,password string                  "+
                        "     ,nickname string                  "+
                        "     ,phone string                     "+
                        "     ,status int                       "+
                        "     ,create_time timestamp(3)         "+
                        "     ,icon string                      "+
                        "     ,gender int                       "+
                        "     ,birthday date                    "+
                        "     ,city string                      "+
                        "     ,job string                       "+
                        "     ,personalized_signature string    "+
                        "     ,source_type int                  "+
                        "     ,integration int                  "+
                        "     ,growth int                       "+
                        "     ,luckey_count int                 "+
                        "     ,history_integration int          "+
                        "     ,modify_time timestamp(3)         "+
                        "     ,primary key(id) not enforced     "+
                        " ) WITH (                              "+
                        "     'connector' = 'mysql-cdc',        "+
                        "     'hostname' = 'doitedu',           "+
                        "     'port' = '3306',                  "+
                        "     'username' = 'root',              "+
                        "     'password' = 'root',              "+
                        "     'database-name' = 'realtimedw',   "+
                        "     'table-name' = 'ums_member'       "+
                        " )                                     "
        );



        // 建表，映射目标表 : hbase中的 dim_user_info
        // hbase> create 'dim_user_info','f'
        tenv.executeSql(
                " create table dim_user_info_hbase(                  "+
                        "      username string                         "+
                        " 	   ,f  ROW<id bigint                       "+
                        "            ,member_level_id bigint           "+
                        "            ,password string                  "+
                        "            ,nickname string                  "+
                        "            ,phone string                     "+
                        "            ,status int                       "+
                        "            ,create_time timestamp(3)         "+
                        "            ,icon string                      "+
                        "            ,gender int                       "+
                        "            ,birthday date                    "+
                        "            ,city string                      "+
                        "            ,job string                       "+
                        "            ,personalized_signature string    "+
                        "            ,source_type int                  "+
                        "            ,integration int                  "+
                        "            ,growth int                       "+
                        "            ,luckey_count int                 "+
                        "            ,history_integration int          "+
                        "            ,modify_time timestamp(3)>        "+
                        "    ,primary key(username) not enforced        "+
                        " ) WITH (                                     "+
                        "  'connector' = 'hbase-2.2',                  "+
                        "  'table-name' = 'dim_user_info',             "+
                        "  'zookeeper.quorum' = 'doitedu:2181'         "+
                        " )                                            "
        );


        // insert  into  select ...
        tenv.executeSql(
                "insert into dim_user_info_hbase " +
                "select username," +
                        "row(id,member_level_id,password,nickname" +
                        ",phone,status,create_time,icon,gender,birthday,city,job," +
                        "personalized_signature,source_type,integration," +
                        "growth,luckey_count,history_integration,modify_time) as f " +
                        "from ums_member_mysql");

    }

}
